//! # Kafka 消息总线适配器示例
//!
//! 本示例展示如何使用 Kafka 适配器进行消息发布和订阅
//!
//! ## 运行前准备
//!
//! 1. 启动 Kafka 服务：
//! ```bash
//! # 启动 Zookeeper
//! bin/zookeeper-server-start.sh config/zookeeper.properties
//! 
//! # 启动 Kafka
//! bin/kafka-server-start.sh config/server.properties
//! ```
//!
//! 2. 创建测试主题（可选）：
//! ```bash
//! bin/kafka-topics.sh --create --topic rustcloud_orders --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
//! bin/kafka-topics.sh --create --topic rustcloud_notifications --bootstrap-server localhost:9092 --partitions 2 --replication-factor 1
//! ```
//!
//! ## 运行示例
//!
//! ```bash
//! cargo run --example kafka_demo --features kafka
//! ```

use rustcloud_bus::{
    kafka_adapter::{KafkaMessageBus, KafkaAdapterConfig},
    bus::MessageBus,
    message::Message,
    error::MessageBusResult,
};
use serde_json::json;
use std::time::Duration;
use tokio::time::sleep;
use tracing::{info, warn, Level};
use uuid::Uuid;

#[tokio::main]
async fn main() -> MessageBusResult<()> {
    // 初始化日志
    tracing_subscriber::fmt()
        .with_max_level(Level::INFO)
        .with_target(false)
        .init();

    info!("🚀 Kafka 消息总线示例开始");

    // 1. 基本用法示例
    basic_usage_example().await?;
    
    // 2. 高级配置示例
    advanced_config_example().await?;
    
    // 3. 多订阅者示例
    multiple_subscribers_example().await?;
    
    // 4. 错误处理示例
    error_handling_example().await?;
    
    // 5. 性能测试示例
    performance_test_example().await?;

    info!("✅ Kafka 消息总线示例完成");
    Ok(())
}

/// 基本用法示例：发布和订阅消息
async fn basic_usage_example() -> MessageBusResult<()> {
    info!("📝 运行基本用法示例");

    // 创建默认配置的 Kafka 消息总线
    let bus = KafkaMessageBus::new().await?;
    
    // 创建订阅者
    let subscriber = bus.subscribe("orders").await?;
    info!("✅ 创建订单订阅者成功");
    
    // 发布消息
    let order_message = Message::new(
        "orders",
        json!({
            "order_id": "ORDER_001",
            "customer_id": "CUST_123",
            "product": "Rust 编程指南",
            "quantity": 2,
            "price": 88.88,
            "status": "pending"
        })
    );
    
    bus.publish(order_message).await?;
    info!("✅ 发布订单消息成功");
    
    // 接收消息
    if let Some(received_msg) = subscriber.receive_timeout(5000).await? {
        info!("📨 接收到订单: {}", received_msg.payload());
        
        // 确认消息
        subscriber.ack(&received_msg).await?;
        info!("✅ 消息确认完成");
    } else {
        warn!("⏰ 未在指定时间内接收到消息");
    }
    
    // 获取统计信息
    let stats = bus.stats().await?;
    info!("📊 消息总线统计: 发布 {}, 接收 {}, 失败 {}", 
          stats.total_published, stats.total_received, stats.total_failed);
    
    // 清理资源
    bus.close().await?;
    info!("🧹 基本用法示例清理完成\n");
    
    Ok(())
}

/// 高级配置示例：自定义 Kafka 配置
async fn advanced_config_example() -> MessageBusResult<()> {
    info!("⚙️ 运行高级配置示例");

    // 创建高级配置
    let config = KafkaAdapterConfig {
        bootstrap_servers: "localhost:9092".to_string(),
        group_id: format!("rustcloud_group_{}", Uuid::new_v4()),
        client_id: format!("rustcloud_client_{}", Uuid::new_v4()),
        session_timeout_ms: 15000,
        heartbeat_interval_ms: 5000,
        enable_auto_commit: false, // 手动提交偏移量
        auto_offset_reset: "earliest".to_string(),
        num_partitions: 5,
        replication_factor: 1,
        batch_size: 32768,
        compression_type: "gzip".to_string(),
        request_timeout_ms: 60000,
        retry_backoff_ms: 200,
        security_protocol: "PLAINTEXT".to_string(),
        sasl_mechanism: None,
        sasl_username: None,
        sasl_password: None,
    };
    
    let bus = KafkaMessageBus::with_config(config).await?;
    info!("✅ 使用高级配置创建 Kafka 消息总线");
    
    // 创建通知订阅者
    let notification_subscriber = bus.subscribe("notifications").await?;
    info!("✅ 创建通知订阅者成功");
    
    // 发布多种类型的通知
    let notifications = vec![
        json!({
            "type": "email",
            "recipient": "user@example.com",
            "subject": "订单确认",
            "content": "您的订单已确认，正在处理中..."
        }),
        json!({
            "type": "sms",
            "phone": "+86-138-0013-8000",
            "content": "【RustCloud】您的订单已发货，请注意查收。"
        }),
        json!({
            "type": "push",
            "device_id": "device_123456",
            "title": "订单更新",
            "content": "您的订单状态已更新为已发货"
        }),
    ];
    
    for (i, notification_data) in notifications.iter().enumerate() {
        let message = Message::new("notifications", notification_data.clone())
            .with_correlation_id(Uuid::new_v4());
        
        bus.publish(message).await?;
        info!("📤 发布通知 {}/{}", i + 1, notifications.len());
    }
    
    // 批量接收通知
    let mut received_count = 0;
    while received_count < 3 {
        if let Some(msg) = notification_subscriber.receive_timeout(3000).await? {
            info!("📨 接收到通知: type={}", 
                  msg.payload().get("type").unwrap_or(&json!("unknown")));
            notification_subscriber.ack(&msg).await?;
            received_count += 1;
        } else {
            break;
        }
    }
    
    info!("📊 成功接收 {} 条通知", received_count);
    
    // 清理资源
    bus.close().await?;
    info!("🧹 高级配置示例清理完成\n");
    
    Ok(())
}

/// 多订阅者示例：展示消费者组的工作方式
async fn multiple_subscribers_example() -> MessageBusResult<()> {
    info!("👥 运行多订阅者示例");

    let bus = KafkaMessageBus::new().await?;
    
    // 创建多个订阅者（同一消费者组）
    let subscriber1 = bus.subscribe("tasks").await?;
    let subscriber2 = bus.subscribe("tasks").await?;
    let subscriber3 = bus.subscribe("tasks").await?;
    
    info!("✅ 创建 3 个任务订阅者（同一消费者组）");
    
    // 发布多个任务
    let tasks = vec![
        json!({"task_id": "TASK_001", "type": "image_processing", "priority": "high"}),
        json!({"task_id": "TASK_002", "type": "data_analysis", "priority": "medium"}),
        json!({"task_id": "TASK_003", "type": "report_generation", "priority": "low"}),
        json!({"task_id": "TASK_004", "type": "backup", "priority": "low"}),
        json!({"task_id": "TASK_005", "type": "cleanup", "priority": "medium"}),
        json!({"task_id": "TASK_006", "type": "monitoring", "priority": "high"}),
    ];
    
    for task_data in &tasks {
        let message = Message::new("tasks", task_data.clone());
        bus.publish(message).await?;
    }
    
    info!("📤 发布 {} 个任务", tasks.len());
    
    // 模拟多个消费者并行处理
    let mut handles = vec![];
    
    // 消费者 1
    let sub1 = subscriber1;
    let handle1 = tokio::spawn(async move {
        let mut count = 0;
        while count < 3 {
            if let Ok(Some(msg)) = sub1.receive_timeout(2000).await {
                let task_id = msg.payload().get("task_id").unwrap();
                info!("🔧 消费者1 处理任务: {}", task_id);
                sleep(Duration::from_millis(100)).await; // 模拟处理时间
                let _ = sub1.ack(&msg).await;
                count += 1;
            } else {
                break;
            }
        }
        count
    });
    
    // 消费者 2
    let sub2 = subscriber2;
    let handle2 = tokio::spawn(async move {
        let mut count = 0;
        while count < 3 {
            if let Ok(Some(msg)) = sub2.receive_timeout(2000).await {
                let task_id = msg.payload().get("task_id").unwrap();
                info!("⚙️ 消费者2 处理任务: {}", task_id);
                sleep(Duration::from_millis(150)).await; // 模拟处理时间
                let _ = sub2.ack(&msg).await;
                count += 1;
            } else {
                break;
            }
        }
        count
    });
    
    // 消费者 3
    let sub3 = subscriber3;
    let handle3 = tokio::spawn(async move {
        let mut count = 0;
        while count < 3 {
            if let Ok(Some(msg)) = sub3.receive_timeout(2000).await {
                let task_id = msg.payload().get("task_id").unwrap();
                info!("🛠️ 消费者3 处理任务: {}", task_id);
                sleep(Duration::from_millis(120)).await; // 模拟处理时间
                let _ = sub3.ack(&msg).await;
                count += 1;
            } else {
                break;
            }
        }
        count
    });
    
    handles.push(handle1);
    handles.push(handle2);
    handles.push(handle3);
    
    // 等待所有消费者完成
    let mut total_processed = 0;
    for handle in handles {
        if let Ok(count) = handle.await {
            total_processed += count;
        }
    }
    
    info!("📊 总计处理任务数: {}", total_processed);
    
    // 清理资源
    bus.close().await?;
    info!("🧹 多订阅者示例清理完成\n");
    
    Ok(())
}

/// 错误处理示例：展示各种错误情况的处理
async fn error_handling_example() -> MessageBusResult<()> {
    info!("❌ 运行错误处理示例");

    // 1. 连接错误处理
    info!("🔍 测试连接错误处理");
    let bad_config = KafkaAdapterConfig {
        bootstrap_servers: "nonexistent:9092".to_string(),
        request_timeout_ms: 5000, // 缩短超时时间
        ..Default::default()
    };
    
    match KafkaMessageBus::with_config(bad_config).await {
        Ok(_) => warn!("⚠️ 预期连接失败，但成功了"),
        Err(e) => info!("✅ 正确捕获连接错误: {}", e),
    }
    
    // 2. 正常配置的消息总线
    let bus = KafkaMessageBus::new().await?;
    
    // 3. 订阅错误处理
    info!("🔍 测试订阅错误处理");
    let subscriber = bus.subscribe("error_test").await?;
    
    // 4. 发布无效消息（这里只是演示，实际不会失败）
    info!("🔍 测试消息发布");
    let valid_message = Message::new("error_test", json!({
        "test": "error_handling",
        "timestamp": chrono::Utc::now().to_rfc3339()
    }));
    
    match bus.publish(valid_message).await {
        Ok(()) => info!("✅ 消息发布成功"),
        Err(e) => warn!("❌ 消息发布失败: {}", e),
    }
    
    // 5. 接收超时处理
    info!("🔍 测试接收超时处理");
    match subscriber.receive_timeout(1000).await {
        Ok(Some(msg)) => {
            info!("📨 接收到消息: {}", msg.payload());
            subscriber.ack(&msg).await?;
        }
        Ok(None) => info!("⏰ 接收超时（正常）"),
        Err(e) => warn!("❌ 接收错误: {}", e),
    }
    
    // 6. 关闭后的操作错误
    info!("🔍 测试关闭后操作错误");
    bus.close().await?;
    
    let closed_message = Message::new("error_test", json!({"after": "close"}));
    match bus.publish(closed_message).await {
        Ok(()) => warn!("⚠️ 预期发布失败，但成功了"),
        Err(e) => info!("✅ 正确捕获关闭后发布错误: {}", e),
    }
    
    info!("🧹 错误处理示例完成\n");
    Ok(())
}

/// 性能测试示例：测试消息吞吐量
async fn performance_test_example() -> MessageBusResult<()> {
    info!("🚀 运行性能测试示例");

    let bus = KafkaMessageBus::new().await?;
    let subscriber = bus.subscribe("performance_test").await?;
    
    let message_count = 1000;
    let start_time = std::time::Instant::now();
    
    // 发布消息
    info!("📤 开始发布 {} 条消息", message_count);
    for i in 0..message_count {
        let message = Message::new(
            "performance_test",
            json!({
                "id": i,
                "data": format!("test_data_{}", i),
                "timestamp": chrono::Utc::now().to_rfc3339()
            })
        );
        
        bus.publish(message).await?;
        
        if i > 0 && i % 100 == 0 {
            info!("📤 已发布 {} 条消息", i);
        }
    }
    
    let publish_duration = start_time.elapsed();
    let publish_rate = message_count as f64 / publish_duration.as_secs_f64();
    
    info!("📊 发布性能: {} 条/秒 (总时间: {:?})", 
          publish_rate as u64, publish_duration);
    
    // 接收消息
    info!("📥 开始接收消息");
    let receive_start = std::time::Instant::now();
    let mut received_count = 0;
    
    while received_count < message_count {
        if let Some(msg) = subscriber.receive_timeout(5000).await? {
            subscriber.ack(&msg).await?;
            received_count += 1;
            
            if received_count % 100 == 0 {
                info!("📥 已接收 {} 条消息", received_count);
            }
        } else {
            warn!("⏰ 接收超时，可能有消息丢失");
            break;
        }
    }
    
    let receive_duration = receive_start.elapsed();
    let receive_rate = received_count as f64 / receive_duration.as_secs_f64();
    
    info!("📊 接收性能: {} 条/秒 (总时间: {:?})", 
          receive_rate as u64, receive_duration);
    
    // 总体统计
    let total_duration = start_time.elapsed();
    let overall_rate = (message_count * 2) as f64 / total_duration.as_secs_f64(); // 发布+接收
    
    info!("📊 总体性能: {} 条/秒 (包含发布和接收)", overall_rate as u64);
    info!("📊 消息成功率: {:.2}%", (received_count as f64 / message_count as f64) * 100.0);
    
    // 获取最终统计
    let final_stats = bus.stats().await?;
    info!("📊 最终统计: 发布 {}, 接收 {}, 失败 {}, 活跃订阅者 {}", 
          final_stats.total_published, 
          final_stats.total_received, 
          final_stats.total_failed,
          final_stats.active_subscribers);
    
    // 清理资源
    bus.close().await?;
    info!("🧹 性能测试示例清理完成\n");
    
    Ok(())
}